package com.bw.gmall.realtime.Week2Sec.service;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 数据处理服务类
 * 用于执行各种数据处理操作
 */
public class DataProcessingService {
    
    private StreamTableEnvironment tableEnv;
    
    public DataProcessingService(StreamTableEnvironment tableEnv) {
        this.tableEnv = tableEnv;
    }
    
    /**
     * 处理新访客数据
     * @param sourceTable 源表名
     * @param targetView 目标视图名
     */
    public void processNewVisitData(String sourceTable, String targetView) {
        String sql = "SELECT " +
                "  common['uid'] as user_id, " +
                "  page['shop'] as shop, " +
                "  page['shop_new_visit'] as shop_new_visit, " +
                "  DATE_FORMAT(rt, 'yyyy-MM-dd') as dt, " +
                "  rt " +
                "FROM " + sourceTable;
        
        Table table = tableEnv.sqlQuery(sql);
        tableEnv.createTemporaryView(targetView, table);
    }
    
    /**
     * 处理新访客订单数据
     * @param visitView 访客视图名
     * @param orderTable 订单表名
     * @param targetView 目标视图名
     */
    public void processNewVisitOrderData(String visitView, String orderTable, String targetView) {
        String sql = "SELECT " +
                "  v.dt as date_id, " +
                "  v.user_id, " +
                "  v.shop " +
                "FROM " + visitView + " v " +
                "JOIN " + orderTable + " o " +
                "  ON v.user_id = o.user_id " +
                "  AND v.shop = o.shop " +
                "  AND v.dt = o.date_id " +
                "WHERE v.shop_new_visit = '1' " +
                "GROUP BY v.dt, v.user_id, v.shop";
        
        Table table = tableEnv.sqlQuery(sql);
        tableEnv.createTemporaryView(targetView, table);
    }
    
    /**
     * 处理未购客户回访数据
     * @param visitView 访客视图名
     * @param orderView 订单视图名
     * @param newOrderView 新订单视图名
     * @param targetView 目标视图名
     */
    public void processNoOrderVisitData(String visitView, String orderView, String newOrderView, String targetView) {
        String sql = "SELECT " +
                "  v.user_id, " +
                "  v.shop, " +
                "  v.dt, " +
                "  v.rt, " +
                "  CAST(NULL AS DOUBLE) as total_amount " +
                "FROM " + visitView + " v " +
                "LEFT JOIN " + orderView + " o " +
                "  ON v.user_id = o.user_id " +
                "  AND v.shop = o.shop " +
                "  AND v.dt = o.date_id " +
                "WHERE CONCAT(v.user_id, '_', v.shop) NOT IN " +
                "  (SELECT CONCAT(user_id, '_', shop) FROM " + newOrderView + ")";
        
        Table table = tableEnv.sqlQuery(sql);
        tableEnv.createTemporaryView(targetView, table);
    }
    
    /**
     * 计算统计指标
     * @param sourceView 源视图名
     * @return 包含统计结果的表
     */
    public Table calculateMetrics(String sourceView) {
        String sql = "SELECT " +
                "  window_start, " +
                "  window_end as date_id, " +
                "  shop as shop_id, " +
                // 1. 未购客户回访总数（去重用户数）
                "  COUNT(DISTINCT user_id) AS new_visit_count, " +
                // 2. 回访成交（有支付金额的去重用户数）
                "  COUNT(DISTINCT CASE WHEN total_amount > 0 THEN user_id END) AS new_visit_paid_count, " +
                // 3. 回访未成交（总回访数 - 成交数）
                "  (COUNT(DISTINCT user_id) - " +
                "   COUNT(DISTINCT CASE WHEN total_amount > 0 THEN user_id END)) AS new_visit_unpaid_count, " +
                // 4. 支付转化率
                "  ROUND(CASE WHEN COUNT(DISTINCT user_id) = 0 THEN 0 " +
                "   ELSE (COUNT(DISTINCT CASE WHEN total_amount > 0 THEN user_id END) * 100.0 " +
                "   / COUNT(DISTINCT user_id)) END, 2) AS payment_conversion_rate, " +
                // 5. 支付总金额
                "  SUM(total_amount) AS total_amount, " +
                // 6. 客单价
                "  ROUND(CASE WHEN COUNT(DISTINCT CASE WHEN total_amount > 0 THEN user_id END) = 0 THEN 0 " +
                "   ELSE (SUM(CASE WHEN total_amount > 0 THEN total_amount END) " +
                "   / COUNT(DISTINCT CASE WHEN total_amount > 0 THEN user_id END)) END, 2) AS customer_price " +
                "FROM TABLE(TUMBLE(TABLE " + sourceView + ", DESCRIPTOR(rt), INTERVAL '10' MINUTES)) " +
                "GROUP BY window_start, window_end, shop";
        
        return tableEnv.sqlQuery(sql);
    }
}